package com.gzhryc.socket.netty;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

import com.gzhryc.common.logging.Logger;
import com.gzhryc.socket.SocketCallback;
import com.gzhryc.socket.SocketConfig;
import com.gzhryc.socket.SocketEvent;
import com.gzhryc.socket.SocketMessage;
import com.gzhryc.socket.SocketNode;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * 字符串TCP服务
 * 
 * @author zhanghao
 *
 */
public class NettyThreadPoolTCPServer extends SocketNode{

	static Logger log = Logger.getLogger(NettyThreadPoolTCPServer.class);

	private ThreadPoolExecutor executor;
	private BlockingQueue<Runnable> blockQueue;
	private MainThread mainThread;
	
	private Channel mChannel;

	Class<? extends ByteNettyChannelHandler> byteNettyChannelHandlerClass;
	SocketCallback mCallback;

	public NettyThreadPoolTCPServer(SocketConfig config,SocketCallback callback) {
		this.config = config;
		this.mCallback = callback;
		
		this.blockQueue = new LinkedBlockingQueue<Runnable>();
    	this.executor = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaximumPoolSize()
    			,config.getKeepAliveTime(),config.getUnit(),this.blockQueue);
	}

	public NettyThreadPoolTCPServer(SocketConfig config,Class<? extends ByteNettyChannelHandler> byteNettyChannelHandlerClass,SocketCallback callback) {
		this.config = config;
		this.mCallback = callback;

		this.blockQueue = new LinkedBlockingQueue<Runnable>();
		this.executor = new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaximumPoolSize()
				,config.getKeepAliveTime(),config.getUnit(),this.blockQueue);
		this.byteNettyChannelHandlerClass = byteNettyChannelHandlerClass;
	}
	
	public void runTask(Runnable runnable) {
    	this.executor.submit(runnable);
    }
	 
	public void start() {
		try {
			final SslContext sslCtx;
			if (config.isSsl()) {
				SelfSignedCertificate ssc = new SelfSignedCertificate();
				sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
			} else {
				sslCtx = null;
			}
			
			final NettyThreadPoolTCPServer self = this;
			EventLoopGroup bossGroup = new NioEventLoopGroup();
			EventLoopGroup workerGroup = new NioEventLoopGroup();
			
			try {
				ServerBootstrap b = new ServerBootstrap();
				b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
						.option(ChannelOption.SO_BACKLOG, 10000).handler(new LoggingHandler(LogLevel.INFO))
						.childHandler(new ChannelInitializer<SocketChannel>() {
							@Override
							public void initChannel(SocketChannel ch) throws Exception {
								ChannelPipeline p = ch.pipeline();
								if (sslCtx != null) {
									p.addLast(sslCtx.newHandler(ch.alloc()));
								}
								if(config.getWaitTime() != null) {
									p.addLast(new IdleStateHandler(config.getWaitTime(), 0, 0, config.getUnit()));
									p.addLast(new AcceptorIdleStateTrigger());
								}
								if(config.getEndSymbol() != null && config.getEndSymbol().length > 0) {
									p.addLast(new DelimiterBasedFrameDecoder(config.getMaxWaitBufferSize(),false,
											Unpooled.copiedBuffer(config.getEndSymbol())));
								}
								p.addLast(new ByteArrayDecoder());
								p.addLast(new ByteArrayEncoder());
								boolean flag = true;
								if(byteNettyChannelHandlerClass != null){
									try {
										Constructor constructor = byteNettyChannelHandlerClass.getConstructor(SocketNode.class,SocketCallback.class);
										ByteNettyChannelHandler byteNettyChannelHandler = ByteNettyChannelHandler.class.cast(constructor.newInstance(self,mCallback));
										p.addLast(byteNettyChannelHandler);
										flag = false;
									} catch (NoSuchMethodException e) {
										log.error(e.getMessage(),e);
									} catch (InvocationTargetException e) {
										log.error(e.getMessage(),e);
									} catch (InstantiationException e) {
										log.error(e.getMessage(),e);
									} catch (IllegalAccessException e) {
										log.error(e.getMessage(),e);
									}
								}

								if(flag){
									p.addLast(new ByteNettyChannelHandler(self, mCallback));
								}
							}
						});
				
				ChannelFuture f = b.bind(config.getPort()).sync();
				mChannel = f.channel();
				
				this.runTask(new Runnable() {
					@Override
					public void run() {
						SocketEvent socketEvent = new SocketEvent(self);
						mCallback.active(socketEvent);
					}
				});
				
				mChannel.closeFuture().sync();
			} finally {
				bossGroup.shutdownGracefully();
				workerGroup.shutdownGracefully();
			}
		} catch (Exception e) {
			log.error(e.getMessage(), e);
		}
	}
	
	public void startService(){
		if(mainThread == null || !mainThread.isAlive()) {
			mainThread = new MainThread(this);
			mainThread.setDaemon(true);		//守护线程
			mainThread.start();
		}
	}

	public static class MainThread extends Thread {
		
		final NettyThreadPoolTCPServer mServer;
		
		public MainThread(NettyThreadPoolTCPServer server) {
			mServer = server;
		}
		
		public void run() {
			mServer.start();
		}
	}

	@Override
	public boolean writeAndFlush(SocketMessage message) {
		return false;
	}

	@Override
	public void close() {
		if(mChannel != null) {
			try {
				mChannel.close().sync();
				final NettyThreadPoolTCPServer self = this;
				this.runTask(new Runnable() {
					@Override
					public void run() {
						SocketEvent socketEvent = new SocketEvent(self);
						mCallback.close(socketEvent);
					}
				});
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public boolean isConnected() {
		if(mChannel != null) {
			return mChannel.isActive();
		}
		return false;
	}

	@Override
	public String getId() {
		if(mChannel != null) {
			return mChannel.id().asLongText();
		}
		return null;
	}

	@Override
	public void reconnect() {
		// TODO Auto-generated method stub
		
	}
	
}
